package cn.edu.xmu.datamining.tangzk.mralgos.sssp;

import java.util.List;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import cn.edu.xmu.datamining.tangzk.util.MyUtil;

public class SingleSourceShortestPathMapper extends
		Mapper<LongWritable, Text, IntWritable, GraphNodeWritable> {

	protected void map(
			LongWritable key,
			Text value,
			org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, IntWritable, GraphNodeWritable>.Context context)
			throws java.io.IOException, InterruptedException {
		// value format: id \t distance: id1:d1, id2:d2, ...
		String val = value.toString();
		String[] flds = val.split("\t");
		if (flds.length < 2) {
			return;
		}
		int nodeId = Integer.parseInt(flds[0]);
		int distance = MyUtil.parseInt(flds[1]);
		String adjString = "";
		if (flds.length >= 3) {
			adjString = flds[2];
		}
		GraphNodeWritable graphNode = new GraphNodeWritable(distance, adjString);
		context.write(new IntWritable(nodeId), graphNode);

		GraphNodeWritable adjNode = null;
		List<GraphEdge> edgeList = graphNode.getEdgeList();
		if (edgeList != null) {
			for (GraphEdge adjEdge : edgeList) {
				if (distance != Integer.MAX_VALUE
						&& adjEdge.getAdjWeight() != Integer.MAX_VALUE) {
					adjNode = new GraphNodeWritable(distance
							+ adjEdge.getAdjWeight());
				} else {
					adjNode = new GraphNodeWritable(Integer.MAX_VALUE);
				}
				context.write(new IntWritable(adjEdge.getNodeId()), adjNode);
			}
		}
	};

}
